package learn.arc;

import io.netty.bootstrap.AbstractBootstrap;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.Channel;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.kqueue.KQueueServerSocketChannel;
import io.netty.channel.nio.AbstractNioChannel;
import io.netty.channel.nio.NioEventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.channel.socket.oio.OioServerSocketChannel;
import io.netty.handler.codec.base64.Base64Decoder;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.Promise;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.CharBuffer;
import java.nio.channels.*;
import java.util.*;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;

/**
 * Channel, Selector, EventLoopGroup/EventLoop
 *  EventLoop包含Selector，定时任务线程池；聚合为EventLoopGroup
 *  Channel通过EventLoopGroup注册到EventLoop selector
 *  serverPort、selector为核心；EventLoopGroup分配线程池，EventLoop执行
 *      单线程单selector；多线程单selector；主从多selector
 *  循环处理请求、定时任务。
 *
 *
 */
public class Netty {
    /**
     *
     */
    public static void basic() throws Exception {
        ServerSocketChannel ssc=null;
        SocketChannel socketChannel;
        Selector selector=null;
        {
            ssc.register(selector, SelectionKey.OP_ACCEPT, null);
            selector.select();
        }
        SelectionKey selectionKey;//

        /* step2，netty；线程池、channel类型、handler*/
        // 启动器
        AbstractBootstrap abstractBootstrap;// parent配置
        ServerBootstrap config;// child配置
        Bootstrap bootstrap;// client
        // 线程池。关联selector，注册、处理channel
        EventLoopGroup elg=null;
        {
            MultithreadEventLoopGroup multithreadEventLoopGroup;
            {
                NioEventLoopGroup nioEventLoopGroup;
                EpollEventLoopGroup eventExecutors;
            }

            EventLoop el=null;// 单线程线程池；构成EventLoopGroup
            SingleThreadEventLoop singleThreadEventLoop;
            {
                NioEventLoop nioEventLoop;// 构造器中新建、绑定selector
                io.netty.channel.epoll.EpollEventLoop epollEventLoop;
            }
        }
        // serverChannel类型。选择selector模型，select、poll、epoll等
        Channel channel;
        {
            io.netty.channel.socket.ServerSocketChannel serverSocketChannel;
            NioServerSocketChannel nioServerSocketChannel;// nio
            EpollServerSocketChannel epollServerSocketChannel;// linux异步
            KQueueServerSocketChannel kQueueServerSocketChannel;// unix异步
            OioServerSocketChannel oioServerSocketChannel;// 阻塞io

            io.netty.channel.socket.SocketChannel socketChannel1;
            NioSocketChannel nioSocketChannel;

            io.netty.bootstrap.ChannelFactory channelFactory;
            ChannelFactory channelFactory1;// 继承前者
            // 反射工厂；无参构造器创建实例
            ReflectiveChannelFactory reflectiveChannelFactory;

        }
        // 频道处理器；编解码、心跳、消息处理等
        ChannelHandler channelHandler;
        {
            ChannelInboundHandler inboundHandler;
            ChannelOutboundHandler outboundHandler;

            ChannelHandlerAdapter handlerAdapter;
            ChannelInboundHandlerAdapter inboundHandlerAdapter;
            ChannelOutboundHandlerAdapter outboundHandlerAdapter;
            ChannelDuplexHandler duplexHandler;

            ChannelHandlerContext handlerContext;
            ChannelPipeline channelPipeline;// handler链
            ChannelInitializer channelInitializer;// 添加inbound handler
            ChannelHandler.Sharable sharable;// 线程间共享，单例模式
        }

        Future nettyFuture;
        {
            // nettyFuture继承javaFuture
            java.util.concurrent.Future javaFuture;

            ChannelFuture channelFuture;// 非阻塞，回调机制
            Promise promise;// 关联channel、eventLoop
            ChannelPromise channelPromise;

        }
    }

    public static void main(String[] args) {
        example();
    }
    /**
     * netty demo；https://blog.csdn.net/weixin_37068368/article/details/80092442
     */
    public static void example(){
        // 我们要创建两个EventLoopGroup，
        // 一个是boss专门用来接收连接，可以理解为处理accept事件，
        // 另一个是worker，可以关注除了accept之外的其它事件，处理子任务。
        //上面注意，boss线程一般设置一个线程，设置多个也只会用到一个，而且多个目前没有应用场景，
        // worker线程通常要根据服务器调优，如果不写默认就是cpu的两倍。
        EventLoopGroup bossGroup=new NioEventLoopGroup();
        EventLoopGroup workerGroup=new NioEventLoopGroup();
        try {
            //服务端要启动，需要创建ServerBootStrap，
            // 在这里面netty把nio的模板式的代码都给封装好了
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup) //配置boss和worker线程
                    // 配置Server的通道，相当于NIO中的ServerSocketChannel
                    .channel(NioServerSocketChannel.class)
                    // parent handler
                    .handler(null)
                    // childHandler表示给worker那些线程配置了一个处理器，
                    // 配置初始化channel，也就是给worker线程配置对应的handler，当收到客户端的请求时，分配给指定的handler处理
                    .childHandler(new ChannelInitializer<io.netty.channel.socket.SocketChannel>() {
                        @Override
                        protected void initChannel(io.netty.channel.socket.SocketChannel socketChannel)
                                throws Exception {
                            socketChannel.pipeline().addLast(new Base64Decoder()); //添加handler，也就是具体的IO事件处理器
                        }
                    });
            //由于默认情况下是NIO异步非阻塞，所以绑定端口后，通过sync()方法阻塞直到连接建立
            // 绑定端口并同步等待客户端连接（sync方法会阻塞，直到整个启动过程完成）
            int port = 8080;
            ChannelFuture channelFuture=bootstrap.bind(port).sync();
            System.out.println("Netty Server Started,Listening on :"+port);
            //等待服务端监听端口关闭
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //释放线程资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    /**
     *
     */
    public static void nioDemo(){
        ServerSocketChannel ssc = null;
        Selector selector = null;
        InetSocketAddress localAddress = new InetSocketAddress(8080);
        Random rnd = new Random();

        try {
            /*创建选择器*/
            selector = Selector.open();

            /*创建服务器通道*/
            ssc = ServerSocketChannel.open();
            ssc.configureBlocking(false);

            /*设置监听服务器的端口，设置最大连接缓冲数为100*/
            ssc.bind(localAddress, 100);

            /*服务器通道只能对tcp链接事件感兴趣*/
            ssc.register(selector, SelectionKey.OP_ACCEPT);

        } catch (IOException e1) {
            System.out.println("server start failed");
            return;
        }

        System.out.println("server start with address : " + localAddress);
        /*服务器线程被中断后会退出*/
        try{

            while(!Thread.currentThread().isInterrupted()){

                int n = selector.select();
                if(n == 0){
                    continue;
                }

                Set<SelectionKey> keySet = selector.selectedKeys();
                Iterator<SelectionKey> it = keySet.iterator();
                SelectionKey key = null;

                while(it.hasNext()){

                    key = it.next();
                    /*防止下次select方法返回已处理过的通道*/
                    it.remove();

                    /*若发现异常，说明客户端连接出现问题,但服务器要保持正常*/
                    try{
                        /*ssc通道只能对链接事件感兴趣*/
                        if(key.isAcceptable()){

                                /*accept方法会返回一个普通通道，
                                     每个通道在内核中都对应一个socket缓冲区*/
                            SocketChannel sc = ssc.accept();
                            sc.configureBlocking(false);

                            /*向选择器注册这个通道和普通通道感兴趣的事件，同时提供这个新通道相关的缓冲区*/
                            int interestSet = SelectionKey.OP_READ;
                            sc.register(selector, interestSet, new ArrayList<Integer>());

                            System.out.println("accept from " + sc.getRemoteAddress());
                        }


                        /*（普通）通道感兴趣读事件且有数据可读*/
                        if(key.isReadable()){

//                            /*通过SelectionKey获取通道对应的缓冲区*/
//                            Object  attachment = key.attachment();
//
//                            /*通过SelectionKey获取对应的通道*/
//                            SocketChannel sc = (SocketChannel) key.channel();
//
//                            /*从底层socket读缓冲区中读入数据*/
//                            sc.read(readBuffer);
//                            readBuffer.flip();
//
//                            /*解码显示，客户端发送来的信息*/
//                            CharBuffer cb = utf8.decode(readBuffer);
//                            System.out.println(cb.array());
//
//                            readBuffer.rewind();
//
//
//                            /*准备好向客户端发送的信息*/
//                            /*先写入"echo:"，再写入收到的信息*/
//                            writeBuffer.put("echo from service:".getBytes("UTF-8"));
//                            writeBuffer.put(readBuffer);
//
//                            readBuffer.clear();
//
//                            /*设置通道写事件*/
//                            key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);

                        }

                        /*通道感兴趣写事件且底层缓冲区有空闲*/
//                        if(key.isWritable()){
//
//                            Buffers  buffers = (Buffers)key.attachment();
//
//                            ByteBuffer writeBuffer = buffers.gerWriteBuffer();
//                            writeBuffer.flip();
//
//                            SocketChannel sc = (SocketChannel) key.channel();
//
//                            int len = 0;
//                            while(writeBuffer.hasRemaining()){
//                                len = sc.write(writeBuffer);
//                                /*说明底层的socket写缓冲已满*/
//                                if(len == 0){
//                                    break;
//                                }
//                            }
//
//                            writeBuffer.compact();
//
//                            /*说明数据全部写入到底层的socket写缓冲区*/
//                            if(len != 0){
//                                /*取消通道的写事件*/
//                                key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE));
//                            }
//
//                        }
                    }catch(IOException e){
                        System.out.println("service encounter client error");
                        /*若客户端连接出现异常，从Seletcor中移除这个key*/
                        key.cancel();
                        key.channel().close();
                    }

                }

                Thread.sleep(rnd.nextInt(500));
            }

        }catch(InterruptedException e){
            System.out.println("serverThread is interrupted");
        } catch (IOException e1) {
            System.out.println("serverThread selecotr error");
        }finally{
            try{
                selector.close();
            }catch(IOException e){
                System.out.println("selector close failed");
            }finally{
                System.out.println("server close");
            }
        }

    }

    /**
     *
     */
    public static void bind(){
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.bind(8080);
        serverBootstrap.bind((InetSocketAddress)null);
        {
            // 1，校验线程池、channelFactory、childHandler
            serverBootstrap.validate();
            // 2，执行绑定
            serverBootstrap.doBind((InetSocketAddress)null);
            {
                // 1，初始化channel，并注册到selector；返回为ChannelPromise对象
                ChannelFuture regFuture = serverBootstrap.initAndRegister();
                {
                    // 1，工厂获取server channel
                    Channel channel = ((ReflectiveChannelFactory)null).newChannel();
                    {
                        // 实例化io.netty.channel.socket.ServerSocketChannel
                        //  1，实例化java.nio.channels.ServerSocketChannel
                        //  2，填充netty serverChannel属性：parentChannel、readInterestOp、blocking等
                        //  3，初始化ServerSocketChannelConfig
                        NioServerSocketChannel serverSocketChannel;

                    }
                    // 2，初始化option、attr、handler
                    serverBootstrap.init((Channel)null);
                    {
                        // 1，设置ChannelOption
                        // 2，设置AttributeKey

                        // 3，channel初始化handler
                        //  serverBootstrap配置的handler
                        //  ServerBootstrapAcceptor
                        ChannelPipeline p = channel.pipeline();
                        // 初始化channel handler，延迟执行。
                        ChannelInitializer channelInitializer;
                        ServerBootstrap.ServerBootstrapAcceptor serverBootstrapAcceptor;// 连接handler，异步添加

                    }
                    // 3，注册到EventLoopGroup
                    regFuture = serverBootstrap.config().group().register(channel);
                    {
                        SingleThreadEventLoop singleThreadEventLoop = new SingleThreadEventLoop();
                        singleThreadEventLoop.register(channel);
                        {
                            // 关联EventLoop、Channel
                            DefaultChannelPromise channelPromise = new DefaultChannelPromise(channel, singleThreadEventLoop);
                            regFuture = singleThreadEventLoop.register(channelPromise);
                            {
                                // netty Future；继承java Future
                                Future nettyFuture;

                                channelPromise.channel().unsafe().register(singleThreadEventLoop, channelPromise);
                                channel.register0(channelPromise);
                                // 执行java的channel.register(selector)
                                // channel绑定到eventLoop的selector
                                ((AbstractNioChannel)channel).doRegister();

                            }
                        }

                    }

                }
                final Channel channel = regFuture.channel();

                // 2，注册完成。
                {
                    // 1，创建新promise
                    // 2，
                    serverBootstrap.doBind0(regFuture, channel, null, null);
                    {
                        // 绑定端口，遍历执行outBoundHandler.bind方法；channelPromise绑定端口绑定完成事件
                        ChannelPromise channelPromise = null;
                        channel.bind(null, channelPromise);
                        // 添加handler
                        channelPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                    }
                }

                // 2，注册未完成。
                // 创建新promise
                // regFuture添加监听器，请求完成时触发

            }
        }


    }

    // https://blog.csdn.net/myth_g/article/details/87877799
    public static class ServerMain {
        public static void run() {
            EventLoopGroup pGroup = new NioEventLoopGroup(); //线程组：用来处理网络事件处理（接受客户端连接）
            EventLoopGroup cGroup = new NioEventLoopGroup(); //线程组：用来进行网络通讯读写
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(pGroup, cGroup)
                        //注册服务端channel
                        .channel(NioServerSocketChannel.class)
                        //队列形式,阻塞式处理队列里的客户端请求
                        // .option(ChannelOption.SO_BACKLOG, 1024)
                        //长连接,每两小时检测心跳
                        .option(ChannelOption.SO_KEEPALIVE, true)
                        //        .handler(new LoggingHandler(LogLevel.INFO))
                        .childHandler(new ChannelInitializer<io.netty.channel.socket.SocketChannel>() {
                            protected void initChannel(io.netty.channel.socket.SocketChannel sc) throws Exception {
                                sc.pipeline()
                                        //添加解码ByteBuf为字符串
                                        .addLast(new StringDecoder())
                                        //编码为字符串,这样发送消息直接使用String即可
                                        //.addLast(new StringEncoder())
                                        .addLast(new ServerHandlerOne());   //服务端业务处理类
                            }
                        });
                ChannelFuture cf = null;
                cf = b.bind(8080).sync();
                cf.channel().closeFuture().sync();
            } catch (Exception e) {
                pGroup.shutdownGracefully();
                cGroup.shutdownGracefully();
                e.printStackTrace();
            }

        }

        public static void main(String[] args) {
            new Thread(
                    () -> {
                        run();
                    }
            ).start();

        }
    }

    public static class ServerHandlerOne extends SimpleChannelInboundHandler<String> {

        private static int count = 0;

        //SimpleChannelInboundHandler带泛型的抽象处理类
        protected void channelRead0(ChannelHandlerContext ctx, String o) throws Exception {
            //避免粘包
            String[] split = o.split("&");
            Arrays.stream(split).forEach((a) -> {
                System.out.println(a);
            });
            //只读的话需要手动释放ByteBuf计数器.
            ReferenceCountUtil.release(o);
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            System.out.println("客户端注册" + count++);
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.out.println("读取客户端信息完成" + count++);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("客户端与服务器连接成功");
            super.channelActive(ctx);
            ByteBuf b = Unpooled.copiedBuffer("我是服务器".getBytes());
            ctx.writeAndFlush(b);
        }
    }

    public static class ClientMain {

        public static void run() {
            EventLoopGroup cGroup = new NioEventLoopGroup(); //线程组：用来进行网络通讯读写
            try {
                Bootstrap b = new Bootstrap();
                b.group(cGroup)
                        .channel(NioSocketChannel.class) //注册服务端channel
                        .option(ChannelOption.SO_KEEPALIVE, true)
                        // .handler(new LoggingHandler(LogLevel.INFO))
                        .handler(new ChannelInitializer<io.netty.channel.socket.SocketChannel>() {
                            protected void initChannel(io.netty.channel.socket.SocketChannel sc) throws Exception {
                                //5s没有交互，就会关闭channel
                                sc.pipeline()
                                        .addLast(new StringDecoder())
                                        .addLast(new ClientHandler())
                                        .addLast(new ClientHandler2());
                            }
                        });
                ChannelFuture localhost = b.connect("localhost", 8080);
                // localhost.addListener(ChannelFutureListener.CLOSE);
                localhost.channel().closeFuture().sync();
                System.out.println("aaaa");
            } catch (Exception e) {
                cGroup.shutdownGracefully();
                e.printStackTrace();
            }

        }

        public static void main(String[] args) {
            run();
        }
    }

    public static class ClientHandler extends SimpleChannelInboundHandler<String> {

        protected void channelRead0(ChannelHandlerContext ctx, String o) throws Exception {
            System.out.println(o);
            //此处使该方法中管道和信息往下传播到下一个处理类
            ctx.fireChannelRead(o);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println(1);
            //此处使该方法中管道和信息往下传播到下一个处理类
            ctx.fireChannelActive();
            new Thread(() -> {
                String s = "客户端&";
                int i = 0;
                do {
                    ByteBuf b = Unpooled.buffer(s.length());
                    try {
                        b.writeBytes(s.getBytes("utf-8"));
                        //注释掉间隔时间,则服务器产生粘包现象
                        // Thread.sleep(500);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    ctx.writeAndFlush(b);
                    i++;
                } while (i != 10);
            }).start();
        }

    }

    public static class ClientHandler2 extends SimpleChannelInboundHandler {

        protected void channelRead0(ChannelHandlerContext ctx, Object o) throws Exception {
            System.out.println(o.toString() + 2);
        }


        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println(2);

        }
    }

}
